package com.example.feature.demo.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper;
import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.kafka.support.converter.StringJsonMessageConverter;
import org.springframework.util.backoff.FixedBackOff;

/**
 * KafkaConfiguration
 *
 * @author zhouxianfeng
 * @date 2021-5-20
 */
@Configuration
public class KafkaConfiguration implements InitializingBean {

    @Autowired
    private KafkaTemplate<Object, Object> template;

/*    @Bean
    public RecordMessageConverter converter() {
        return new StringJsonMessageConverter();
    }

    @Bean
    public BatchMessagingMessageConverter batchConverter() {
        return new BatchMessagingMessageConverter(converter());
    }*/

    @Bean
    public SeekToCurrentErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
        return new SeekToCurrentErrorHandler(
                new DeadLetterPublishingRecoverer(template), new FixedBackOff(1000L, 2));
    }

    @Bean
    public RecordMessageConverter converter() {
        StringJsonMessageConverter converter = new StringJsonMessageConverter();
        DefaultJackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper();
        typeMapper.setTypePrecedence(Jackson2JavaTypeMapper.TypePrecedence.TYPE_ID);
        typeMapper.addTrustedPackages("com.example.feature.demo.dto", "com.example.feature.demo.entity");
        /*Map<String, Class<?>> mappings = new HashMap<>();
        mappings.put("foo", Foo2.class);
        mappings.put("bar", Bar2.class);
        typeMapper.setIdClassMapping(mappings);*/
        converter.setTypeMapper(typeMapper);
        return converter;
    }

    @Bean
    public NewTopic topic1() {
        return TopicBuilder.name("topic-zhou1").partitions(3).replicas(3).build();
    }

    @Bean
    public NewTopic topic2() {
        return TopicBuilder.name("topic-zhou2").partitions(3).replicas(3).build();
    }

    @Bean
    public NewTopic topic3() {
        return TopicBuilder.name("topic-zhou3").partitions(3).replicas(3).build();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        template.setAllowNonTransactional(true);
    }
}
